package com.rem.rabbitmq.ee.Fconfirm;

import com.rabbitmq.client.Channel;
import com.rem.rabbitmq.ee.RabbitMqUtils;


/**
 * 发布确认
 * 1.要求队列必须持久胡
 * 2.要求队列里的消息必须持久化
 * 3.发布确认
 * <p>
 * <p>
 * 发布确认原理
 * 所有在该信道上面发布的消息都将会被指派一个唯一的ID，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，
 * 这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出
 * <p>
 * 批量确认  每发100个确认一次  这种方案仍然是同步的，也一样阻塞消息的发布
 * 批量同步等待确认，简单，合理的吞吐量，一旦出现问题但很难推断出是哪条消息出现了问题。
 *
 * @author Rem
 * @date 2021-12-24
 */

public class BatchConfirmProducer {

    private final static int MESSAGE_COUNT = 1005;

    public static void main(String[] args) throws Exception {

        try (Channel channel = RabbitMqUtils.getChannel();) {

            //声明队列
            channel.queueDeclare(RabbitMqUtils.QUEUE_CONFIRM, true, false, false, null);
            /**
             * 启用发布确认
             */
            channel.confirmSelect();

            System.err.println("开始发送消息...");
            //批量确认消息大小
            int batchSize = 100;
            //未确认消息个数
            int outstandingMessageCount = 0;
            long begin = System.currentTimeMillis();

            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + ":confirm";
                channel.basicPublish("", RabbitMqUtils.QUEUE_CONFIRM, null, message.getBytes());
                /**
                 * 等到自上次调用以来发布的所有消息都被代理确认或确认。请注意，在非确认通道上调用时，waitForConfirms 会抛出 IllegalStateException。
                 *
                 * return 是否所有消息都被确认（并且没有消息被确认）服务端返回false或超时时间内未返回，生产者可以消息重发
                 */
                outstandingMessageCount++;
                if (outstandingMessageCount == batchSize) {
                    channel.waitForConfirms();
                    outstandingMessageCount = 0;
                    System.out.println(i + "消息批量" + batchSize + "发送成功");

                }
            }

            /**
             * 最后未被确认的 再次确认
             */
            if (outstandingMessageCount > 0) {
                channel.waitForConfirms();
                System.out.println("last 消息发送成功");
            }

            long end = System.currentTimeMillis();
            System.err.println("发布" + MESSAGE_COUNT + "个批量确认消息,耗时" + (end - begin) + "ms");
        } finally {
            RabbitMqUtils.close();
        }

    }
}
